package com.udf.flink.scala.Catalogs

import java.util
import java.util.Optional

import org.apache.flink.connector.jdbc.catalog.JdbcCatalog
import org.apache.flink.table.api.{EnvironmentSettings, TableEnvironment}
import org.apache.flink.table.catalog.CatalogBaseTable
import org.apache.flink.table.catalog.CatalogBaseTable.TableKind
//import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.table.catalog.{CatalogDatabaseImpl, ObjectPath}

object Pg {
  def main(args: Array[String]): Unit = {
    val bsSettings = EnvironmentSettings.newInstance().inBatchMode().build()
    val t_env = TableEnvironment.create(bsSettings)
    val name = "dwh"
    val default_database = "dwh"
    val username = "dcetl"
    val password = "1q2w3e4r@dcetl"
    val base_url = "jdbc:postgresql://192.168.58.101:5432/"
//    val pg =new JdbcCatalog(name, default_database, username, password, base_url)
    var catalog_sql=s"""CREATE CATALOG $name  WITH(
                      |    'type' = 'jdbc',
                      |    'default-database' = 'dwh',
                      |    'username' = 'dcetl',
                      |    'password' = '$password',
                      |    'base-url' = '$base_url'
                      |)""".stripMargin
    t_env.executeSql(catalog_sql)
//    t_env.registerCatalog("dwh", pg)
//    pg.createDatabase("demo",new CatalogDatabaseImpl())
    val pg=  t_env.getCatalog("dwh").get()
    println(pg.listDatabases())
//    println(pg.listTables("dwh"))
//    pg.createTable(new ObjectPath("dwh","t2"),new CatalogBaseTable {
//      override def getTableKind: CatalogBaseTable.TableKind = TableKind.TABLE
//
//      override def getOptions: util.Map[String, String] = ???
//
//      override def getComment: String = ???
//
//      override def copy(): CatalogBaseTable = ???
//
//      override def getDescription: Optional[String] = ???
//
//      override def getDetailedDescription: Optional[String] = ???
//    })
    val df=pg.getTable(new ObjectPath("dwh","edw.dim_cust"))
//    println(df.getDescription(),df.getComment,df.getOptions,df.getDetailedDescription)
    t_env.useCatalog("dwh")
    t_env.useDatabase("dwh")
    val cust=t_env.from("`edw.dim_cust`")
//    println(cust.printSchema())
//    println(cust.getResolvedSchema)
//    print( cust.explain())
//    cust.limit(50).execute().print()
    t_env.sqlQuery("select cust_cd,concat_ws('^',left(cust_nm,2),cust_cd) as ts from `edw.dim_cust` where cert_type_cd='01' ").execute().print()





  }



}